package com.atguigu.gmall.realtime.dim.app;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.common.bean.TableProcessDim;
import com.atguigu.gmall.realtime.common.constant.Constant;
import com.atguigu.gmall.realtime.common.util.HBaseUtil;
import com.atguigu.gmall.realtime.common.util.JdbcUtil;
import com.atguigu.gmall.realtime.common.util.RedisUtil;
import com.atguigu.gmall.realtime.dim.function.HBaseSinkFunction;
import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import com.ververica.cdc.connectors.mysql.table.StartupOptions;
import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.serialization.DeserializationSchema;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.runtime.state.hashmap.HashMapStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.BroadcastStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.apache.flink.util.Collector;
import org.apache.hadoop.hbase.client.Connection;
import redis.clients.jedis.Jedis;

import java.io.IOException;
import java.util.*;


/**
 * @program: gmall2024-realtime-dev
 * @description: dim层数据同步, 没有抽取代码之前的样子
 * @author: GG-lyf
 * @create: 2024-10-01 21:15:02
 */
@SuppressWarnings("all")
public class DimTest {
    public static void main(String... args) throws Exception {
        // TODO 配置环境
        String ckAndGroupId = "dim_app";
        //Configuration configuration = new Configuration();
        //configuration.setInteger("rest.port", 10002);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(4);
        env.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointTimeout(60000L);
        env.getCheckpointConfig().setExternalizedCheckpointCleanup(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(2000L);
        env.getCheckpointConfig().setCheckpointStorage("hdfs://192.168.123.129:8020/ck");
        env.setRestartStrategy(RestartStrategies.failureRateRestart(3, Time.days(30), Time.seconds(30)));
        env.setStateBackend(new HashMapStateBackend());

        System.setProperty("HADOOP_USER_NAME", "root");
        // TODO: 从kafka中读取maxwell写进topic_db这个主题的数据
        KafkaSource<String> kafkaSource = KafkaSource.<String>builder()
                .setBootstrapServers("192.168.123.129:9092")
                .setTopics("topic_db")
                .setGroupId(ckAndGroupId)
                .setStartingOffsets(OffsetsInitializer.latest())
                .setValueOnlyDeserializer(
                        new DeserializationSchema<String>() {
                            @Override
                            public TypeInformation<String> getProducedType() {
                                return TypeInformation.of(String.class);
                            }

                            @Override
                            public String deserialize(byte[] message) throws IOException {
                                return message == null ? null : new String(message);
                            }

                            @Override
                            public boolean isEndOfStream(String s) {
                                return false;
                            }
                        }
                )
                .build();

        DataStreamSource<String> kafkaStrDS = env.fromSource(kafkaSource, WatermarkStrategy.noWatermarks(), "kafka-source");
        //TODO 把Kafka中的业务数据进行过滤,找到需要的表数据
        SingleOutputStreamOperator<JSONObject> jsonObjDS = kafkaStrDS.process(
                new ProcessFunction<String, JSONObject>() {
                    /**
                     * 处理kafka中的数据
                     * @param s 源数据
                     * @param context 上下文
                     * @param collector 收集器,往下游channel中输出数据
                     * @throws Exception
                     */
                    @Override
                    public void processElement(String s, ProcessFunction<String, JSONObject>.Context context, Collector<JSONObject> out) throws Exception {
                        JSONObject object = JSONObject.parseObject(s);
                        String database = object.getString("database");
                        String type = object.getString("type");
                        String data = object.getString("data");
                        if ("gmall2024".equals(database) && ("insert".equals(type) || "update".equals(type) || "delete".equals(type) || "bootstrap-insert".equals(type)) && data != null && data.length() > 2) {
                            out.collect(object);
                        }
                    }
                }
        );
        //jsonObjDS.print();

        //TODO 使用flink-cdc读取配置表(维度表)
        Properties props = new Properties();
        props.setProperty("useSSL", "false");
        props.setProperty("allowPublicKeyRetrieval", "true");
        MySqlSource<String> mySqlSource = MySqlSource.<String>builder()
                .hostname("192.168.123.129")
                .port(3306)
                .username("root")
                .password("000000")
                .databaseList("gmall2024_config")
                .tableList("gmall2024_config" + "." + "table_process_dim")
                .deserializer(new JsonDebeziumDeserializationSchema())
                .startupOptions(StartupOptions.initial())
                .jdbcProperties(props)
                .build();
        DataStreamSource<String> mysqlStrDS = env.fromSource(mySqlSource, WatermarkStrategy.noWatermarks(), "mysql_source").setParallelism(1);
        //mysqlStrDS.print();
        // TODO 维度表数据进行数据类型转换 jsonStr->实体类对象
        SingleOutputStreamOperator<TableProcessDim> tpDS = mysqlStrDS.map(
                new MapFunction<String, TableProcessDim>() {

                    @Override
                    public TableProcessDim map(String s) throws Exception {
                        JSONObject object = JSONObject.parseObject(s);
                        TableProcessDim tableProcessDim = new TableProcessDim();
                        String op = object.getString("op");
                        if ("d".equals(op)) {
                            //对配置表进行了一次删除操作   从before属性中获取删除前的配置信息
                            tableProcessDim = object.getObject("before", TableProcessDim.class);
                        } else {
                            //对配置表进行了读取、添加、修改操作   从after属性中获取最新的配置信息
                            tableProcessDim = object.getObject("after", TableProcessDim.class);
                        }
                        tableProcessDim.setOp(op);
                        return tableProcessDim;
                    }
                }
        ).setParallelism(1);
        //tpDS.print();

        //TODO 根据配置表中的配置信息到HBase中执行建表或者删除表操作
        tpDS = tpDS.map(
                new RichMapFunction<TableProcessDim, TableProcessDim>() {
                    private Connection hbaseConn;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        hbaseConn = HBaseUtil.getHBaseConnection();
                    }

                    @Override
                    public void close() throws Exception {
                        HBaseUtil.closeHBaseConnection(hbaseConn);
                    }

                    @Override
                    public TableProcessDim map(TableProcessDim tp) throws Exception {
                        String op = tp.getOp();
                        //获取Hbase中维度表的表名
                        String sinkTable = tp.getSinkTable();
                        //获取在HBase中建表的列族
                        String[] sinkFamilies = tp.getSinkFamily().split(",");
                        if ("d".equals(op)) {
                            //从配置表中删除了一条数据  将hbase中对应的表删除掉
                            HBaseUtil.dropHBaseTable(hbaseConn, "gmall2024", sinkTable);
                        } else if ("r".equals(op) || "c".equals(op)) {
                            //从配置表中读取了一条数据或者向配置表中添加了一条配置   在hbase中执行建表
                            HBaseUtil.createHBaseTable(hbaseConn, "gmall2024", sinkTable, sinkFamilies);
                        } else {
                            //对配置表中的配置信息进行了修改   先从hbase中将对应的表删除掉，再创建新表
                            HBaseUtil.dropHBaseTable(hbaseConn, "gmall2024", sinkTable);
                            HBaseUtil.createHBaseTable(hbaseConn, "gmall2024", sinkTable, sinkFamilies);
                        }
                        return tp;
                    }
                }
        ).setParallelism(1);
        //tpDS.print();

        //TODO 从业务数据中过滤出维度数据
        //维度表广播
        MapStateDescriptor<String, TableProcessDim> mapStateDescriptor = new MapStateDescriptor<>("mapStateDescriptor", String.class, TableProcessDim.class);
        BroadcastStream<TableProcessDim> broadcastDS = tpDS.broadcast(mapStateDescriptor);
        SingleOutputStreamOperator<Tuple2<JSONObject, TableProcessDim>> dimDS = jsonObjDS.connect(broadcastDS)
                .process(
                        new BroadcastProcessFunction<JSONObject, TableProcessDim, Tuple2<JSONObject, TableProcessDim>>() {
                            private Map<String, TableProcessDim> configMap = new HashMap<>();

                            @Override
                            public void open(Configuration parameters) throws Exception {
                                //将配置表中的配置信息预加载到程序configMap中,怕后面用到时读出空
                                java.sql.Connection mySQLConnection = JdbcUtil.getMySQLConnection();
                                List<TableProcessDim> tableProcessDimList = JdbcUtil.queryList(mySQLConnection, "select * from gmall2024_config.table_process_dim", TableProcessDim.class, true);
                                for (TableProcessDim tableProcessDim : tableProcessDimList) {
                                    configMap.put(tableProcessDim.getSourceTable(), tableProcessDim);
                                }
                                JdbcUtil.closeMySQLConnection(mySQLConnection);
                            }

                            /**
                             * 处理主流数据
                             * @param jsonObj
                             * @param ctx
                             * @param out
                             * @throws Exception
                             */
                            @Override
                            public void processElement(JSONObject jsonObj, BroadcastProcessFunction<JSONObject, TableProcessDim, Tuple2<JSONObject, TableProcessDim>>.ReadOnlyContext ctx, Collector<Tuple2<JSONObject, TableProcessDim>> out) throws Exception {
                                //获取处理的数据的表名
                                String table = jsonObj.getString("table");
                                //获取广播状态
                                ReadOnlyBroadcastState<String, TableProcessDim> broadcastState = ctx.getBroadcastState(mapStateDescriptor);
                                //根据表名先到广播状态中获取对应的配置信息，如果没有找到对应的配置，再尝试到configMap中获取
                                TableProcessDim tableProcessDim = null;

                                if ((tableProcessDim = broadcastState.get(table)) != null || (tableProcessDim = configMap.get(table)) != null) {
                                    //如果根据表名获取到了对应的配置信息，说明当前处理的是维度数据
                                    // 将维度数据继续向下游传递(只需要传递data属性内容即可)
                                    JSONObject dataJsonObj = jsonObj.getJSONObject("data");
                                    //在向下游传递数据前，过滤掉不需要传递的属性
                                    String sinkColumns = tableProcessDim.getSinkColumns();
                                    deleteNotNeedColumns(dataJsonObj, sinkColumns);
                                    //在向下游传递数据前，补充对维度数据的操作类型属性
                                    String type = jsonObj.getString("type");
                                    dataJsonObj.put("type", type);
                                    out.collect(Tuple2.of(dataJsonObj, tableProcessDim));
                                }
                            }

                            /**
                             * 处理广播数据 将配置数据放到广播状态中或者从广播状态中删除对应的配置信息 key: 维度表名称 value: 配置信息
                             * @param tp
                             * @param ctx
                             * @param collector
                             * @throws Exception
                             */
                            @Override
                            public void processBroadcastElement(TableProcessDim tp, BroadcastProcessFunction<JSONObject, TableProcessDim, Tuple2<JSONObject, TableProcessDim>>.Context ctx, Collector<Tuple2<JSONObject, TableProcessDim>> collector) throws Exception {
                                //获取对配置表进行的操作的类型
                                String op = tp.getOp();
                                //获取广播状态
                                BroadcastState<String, TableProcessDim> broadcastState = ctx.getBroadcastState(mapStateDescriptor);
                                //获取维度表名称
                                String sourceTable = tp.getSourceTable();
                                if ("d".equals(op)) {
                                    //从配置表中删除了一条数据，将对应的配置信息也从广播状态中删除
                                    broadcastState.remove(sourceTable);
                                    configMap.remove(sourceTable);
                                } else {
                                    //对配置表进行了读取、添加或者更新操作，将最新的配置信息放到广播状态中
                                    broadcastState.put(sourceTable, tp);
                                    configMap.put(sourceTable, tp);
                                }
                            }

                            private void deleteNotNeedColumns(JSONObject dataJsonObj, String sinkColumns) {
                                List<String> columnList = Arrays.asList(sinkColumns.split(","));
                                Set<Map.Entry<String, Object>> entrySet = dataJsonObj.entrySet();
                                entrySet.removeIf(entry -> !columnList.contains(entry.getKey()));
                            }
                        }
                );
        //dimDS.print();

        //TODO 维度数据写入HBase 由于没有 HBaseSinkFunction 类，所以只能使用旧的addSink方法自己实现插入过程,如果有 HBaseSinkFunction 类，可以直接使用sinkTo方法
        dimDS.addSink(new RichSinkFunction<Tuple2<JSONObject, TableProcessDim>>() {
            private Connection hbaseConn;
            @Override
            public void open(Configuration parameters) throws Exception {
                hbaseConn = HBaseUtil.getHBaseConnection();
            }
            @Override
            public void close() throws Exception {
                HBaseUtil.closeHBaseConnection(hbaseConn);
            }

            @Override
            public void invoke(Tuple2<JSONObject, TableProcessDim> tup, Context context) throws Exception {
                JSONObject jsonObj = tup.f0;
                TableProcessDim tableProcessDim = tup.f1;
                String type = jsonObj.getString("type");
                jsonObj.remove("type");

                //获取操作的HBase表的表名
                String sinkTable = tableProcessDim.getSinkTable();
                //获取rowkey
                String rowKey = jsonObj.getString(tableProcessDim.getSinkRowKey());
                //判断对业务数据库维度表进行了什么操作
                if ("delete".equals(type)) {
                    //从业务数据库维度表中做了删除操作  需要将HBase维度表中对应的记录也删除掉
                    HBaseUtil.delRow(hbaseConn, Constant.HBASE_NAMESPACE, sinkTable, rowKey);
                } else {
                    //如果不是delete，可能的类型有insert、update、bootstrap-insert，上述操作对应的都是向HBase表中put数据
                    String sinkFamily = tableProcessDim.getSinkFamily();
                    HBaseUtil.putRow(hbaseConn, Constant.HBASE_NAMESPACE, sinkTable, rowKey, sinkFamily, jsonObj);
                }
            }
        });
        dimDS.print();

        env.execute();
    }
}
